#include "threadpool.h"
#include <iostream>
#include <cstring>
#include <unistd.h>

#define DEFAULT_MIN_THREADS 3
#define DEFAULT_MAX_THREADS 8
#define DEFAULT_MAX_QUEUE_CAPACITY 20

#define SLEEP_INTERVAL 5
#define INCREASE_NUM 2
#define DECREASE_NUM 2

// 管理者线程
void *managerThreadFunc(void *arg)
{
    // 线程分离
    pthread_detach(pthread_self());

    ThreadPool *pool = (ThreadPool *)arg;

    int ret = 0;
    // 调度
    while (true)
    {
        // 休眠5s
        sleep(SLEEP_INTERVAL);

        // 加锁
        pthread_mutex_lock(&pool->m_mutex);
        int aliveThreadNum = pool->m_aliveThreadNum;
        int queueSize = pool->m_queueSize;
        // 解锁
        pthread_mutex_unlock(&pool->m_mutex);

        // 加锁
        pthread_mutex_lock(&(pool->m_busyMutex));
        int busyThreadNum = pool->m_busyThreadNum;
        // 解锁
        pthread_mutex_unlock(&pool->m_busyMutex);

        std::cout << "queueSize:" << queueSize << ", aliveThreadNum:" << aliveThreadNum << ", busyThreadNum:" << busyThreadNum << std::endl;

        // 1、什么时候需要增加线程呢？
        if (queueSize > (aliveThreadNum - busyThreadNum) && aliveThreadNum < pool->m_maxThreadNum)
        {
            int addCnt = 0;
            // 加锁
            pthread_mutex_lock(&(pool->m_mutex));

            for (int idx = 0; idx < pool->m_maxThreadNum && addCnt < INCREASE_NUM; idx++)
            {
                // 探测
                if (pool->m_threadId[idx] == 0)
                {
                    ret = pthread_create(&(pool->m_threadId[idx]), NULL, workerThreadFunc, pool);
                    if (ret == -1)
                    {
                        perror("thread create error");
                        _exit(-1);
                    }
                    printf("%lu join thread pool.\n", pool->m_threadId[idx]);

                    addCnt++;
                    // 存活的线程数++
                    (pool->m_aliveThreadNum)++;
                }
            }

            // 解锁
            pthread_mutex_unlock(&(pool->m_mutex));
        }

        // 2、什么时候减少线程池的线程呢？
        if (pool->m_queueSize < aliveThreadNum - busyThreadNum && aliveThreadNum > pool->m_minThreadNum)
        {
            // 减少线程
            pthread_mutex_lock(&(pool->m_mutex));

            // 减少的线程数
            pool->exitNum = DECREASE_NUM;

            for (int idx = 0; idx < DECREASE_NUM; idx++)
            {
                pthread_cond_signal(&(pool->m_queueNotEmpty));
            }

            // 解锁
            pthread_mutex_unlock(&(pool->m_mutex));
        }
    }
}

// 线程工作函数
void *workerThreadFunc(void *arg)
{
    // 线程分离
    pthread_detach(pthread_self());

    ThreadPool *pool = (ThreadPool *)arg; // 将 void* 类型的参数 arg 强转为 ThreadPool* 类型的指针
    while (true)
    {
        // 阶段 1: 等待任务进入队列
        pthread_mutex_lock(&(pool->m_mutex)); // 加锁保护队列
        // 如果任务队列为空，等待任务到来或退出信号
        while (pool->m_queueSize == 0)
        {
            usleep(10000); // 休眠10ms

            // 等待任务队列非空信号
            // 这个函数三个作用：1、阻塞等待（让出CPU） 2、解锁 3、加锁
            pthread_cond_wait(&(pool->m_queueNotEmpty), &(pool->m_mutex));

            // 检查是否有线程退出请求
            if (pool->exitNum > 0)
            {
                pool->exitNum--;
                if (pool->m_aliveThreadNum > pool->m_minThreadNum)
                {
                    // 存活的线程数减一
                    pool->m_aliveThreadNum--;
                    pthread_mutex_unlock(&(pool->m_mutex)); // 解锁
                    printf("%lu exit thread pool...\n", pthread_self()); // 离开了线程池

                    // 线程退出
                    pool->exitThread(pthread_self());
                }
            }
        }
        // 程序到这个地方说明：可以从任务队列里拿数据

        // 阶段 2: 任务出队
        Task_t newTask = pool->m_queue[pool->m_queueFront];
        pool->m_queueFront = (pool->m_queueFront + 1) % pool->m_queueCapacity; // 将队列的队头向后迭代
        pool->m_queueSize--; // 队列的任务个数减一
        pthread_mutex_unlock(&(pool->m_mutex)); // 解锁队列锁

        // 更新忙碌线程数
        pthread_mutex_lock(&(pool->m_busyMutex));
        pool->m_busyThreadNum++; // 忙碌的线程数加一
        pthread_mutex_unlock(&(pool->m_busyMutex));

        // 线程开始执行任务
        newTask.function(newTask.arg);

        // 阶段 3: 任务完成后的清理
        pthread_mutex_lock(&(pool->m_busyMutex));
        pool->m_busyThreadNum--; // 忙碌的线程数减一
        pthread_mutex_unlock(&(pool->m_busyMutex));

        // 发送任务队列非满的信号，通知等待的生产者线程：队列有空间了
        pthread_cond_signal(&pool->m_queueNotFull);
    }
}

// 构造函数
ThreadPool::ThreadPool(int minThread, int maxThread, int maxQueueCapacity)
{
    if (minThread < 0 || maxThread <= 0 || minThread >= maxThread)
    {
        minThread = DEFAULT_MIN_THREADS;
        maxThread = DEFAULT_MAX_THREADS;
    }

    if (maxQueueCapacity <= 0)
    {
        maxQueueCapacity = DEFAULT_MAX_QUEUE_CAPACITY;
    }

    // 队列
    this->m_queueCapacity = maxQueueCapacity;
    // 队列的队头和队尾
    this->m_queueFront = this->m_queueTail = 0;
    // 队列的元素个数
    this->m_queueSize = 0;
    // 任务队列
    this->m_queue = new Task_t[maxQueueCapacity];
    // 清除脏数据
    memset(this->m_queue, 0, sizeof(Task_t) * maxQueueCapacity);

    // 工作线程Id
    m_threadId = new pthread_t[maxThread];
    // 清除脏数据
    memset(m_threadId, 0, sizeof(pthread_t) * maxThread);

    this->m_minThreadNum = minThread;
    this->m_maxThreadNum = maxThread;
    this->m_aliveThreadNum = 0;
    this->m_busyThreadNum = 0;

    int ret = 0;
    // 创建管理者线程
    ret = pthread_create(&m_managerThreadId, NULL, managerThreadFunc, this);
    if (ret == -1)
    {
        perror("thread create error");
        _exit(-1);
    }

    // 创建工作线程
    for (int idx = 0; idx < minThread; idx++)
    {
        ret = pthread_create(&m_threadId[idx], NULL, workerThreadFunc, this);
        if (ret == -1)
        {
            perror("thread create error");
            _exit(-1);
        }
        // 存活的线程函数++
        (this->m_aliveThreadNum)++;
    }

    // 初始化锁
    pthread_mutex_init(&m_mutex, NULL);
    pthread_mutex_init(&m_busyMutex, NULL);
    // 初始化条件变量
    pthread_cond_init(&m_queueNotEmpty, NULL);
    pthread_cond_init(&m_queueNotFull, NULL);

    this->exitNum = 0;
}

// 析构函数
ThreadPool::~ThreadPool()
{
    if (m_threadId != NULL)
    {
        delete[] m_threadId;
    }

    if (m_queue != NULL)
    {
        delete[] m_queue;
    }

    pthread_mutex_destroy(&m_mutex);
    pthread_mutex_destroy(&m_busyMutex);

    pthread_cond_destroy(&m_queueNotEmpty);
    pthread_cond_destroy(&m_queueNotFull);
}

// 添加任务
void ThreadPool::addTask(void *(*function)(void *arg), void *arg)
{
    // 将任务添加到任务队列

    // 加锁
    pthread_mutex_lock(&m_mutex);

    while (this->m_queueCapacity == this->m_queueSize)
    {
        pthread_cond_wait(&m_queueNotFull, &m_mutex);
    }
    // 程序执行到这里：将任务添加到队列里面
    m_queue[m_queueTail].function = function;
    m_queue[m_queueTail].arg = arg;

    m_queueTail = (m_queueTail + 1) % m_queueCapacity;

    // 任务数量加一
    this->m_queueSize++;

    // 解锁
    pthread_mutex_unlock(&m_mutex);

    // 发信号
    pthread_cond_signal(&m_queueNotEmpty);
}

// 线程退出
void ThreadPool::exitThread(pthread_t tid)
{
    for (int idx = 0; idx < this->m_maxThreadNum; idx++)
    {
        if (m_threadId[idx] == tid)
        {
            m_threadId[idx] = 0;

            // 线程退出
            pthread_exit(NULL);
        }
    }
}
